import importlib
import json
import os
import socket
import time
import traceback
from multiprocessing import Process

from air_web.config.config import config
from air_web.dw import gen_task_log
from air_web.dw.data_mapping import StatusCode
from air_web.dw.dws_15min import AggMain
from air_web.dw.spec_cons_15min import SpecConsAggMain

task_base_save_table = config.get("ORGNO_TYPEID_15MIN", "orgno_typeid_15min")
task_baseline_save_table = config.get(
    "BASELINE_ORGNO_TYPEID_15MIN", "baseline_orgno_typeid_15min"
)
orderly_power_15min = config.get("ORDERLY_POWER_15MIN", "orderly_power_15min")
demand_power_15min = config.get("DEMAND_POWER_15MIN", "demand_power_15min")

task_first = {
    "step_init": {
        "type": "init",
        "module": "air_web.dw.dws_first",
        "class": "InitJudge",
        "is_valid": 1,
    },
    "step_cons": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "CCons",
        "save_table": "c_cons",
        "is_valid": 1,
    },
    "step_merge_cons": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "MergeCons",
        "save_table": "merge_cons",
        "is_valid": 1,
    },
    "step_area_type_map": {
        "type": "proc_cons",
        "module": "air_web.dw.dws_first",
        "class": "AreaTypeMap",
        "save_table": "area_type_map",
        "is_valid": 1,
    },
}
task_base = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.dws_15min",
        "class": "AggBase",
        "is_baseline": False,
    },
    "step_on7_type": {
        "type": "agg",
        "dimensions": ["on7", "type_id", "data_time"],
        "rename": {"on7": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_pare": {
        "type": "agg",
        "dimensions": ["on7", "pare_type_id", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_sort": {
        "type": "agg",
        "dimensions": ["on7", "type_code_sort", "data_time"],
        "rename": {"on7": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_all": {
        "type": "agg",
        "dimensions": ["on7", "highest_type_id", "data_time"],
        "rename": {"on7": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_type": {
        "type": "agg",
        "dimensions": ["on5", "type_id", "data_time"],
        "rename": {"on5": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "agg",
        "dimensions": ["on5", "pare_type_id", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_sort": {
        "type": "agg",
        "dimensions": ["on5", "type_code_sort", "data_time"],
        "rename": {"on5": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_all": {
        "type": "agg",
        "dimensions": ["on5", "highest_type_id", "data_time"],
        "rename": {"on5": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_type": {
        "type": "agg",
        "dimensions": ["province_no", "type_id", "data_time"],
        "rename": {"province_no": "org_no"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "agg",
        "dimensions": ["province_no", "pare_type_id", "data_time"],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_sort": {
        "type": "agg",
        "dimensions": ["province_no", "type_code_sort", "data_time"],
        "rename": {"province_no": "org_no", "type_code_sort": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_all": {
        "type": "agg",
        "dimensions": ["province_no", "highest_type_id", "data_time"],
        "rename": {"province_no": "org_no", "highest_type_id": "type_id"},
        "save_table": task_base_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
}
task_industry_agg = {
    "step_base": {
        "type": "industry_base_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAggBase",
        "source_table": task_base_save_table,
    },
    "step_industry_agg": {
        "type": "step_industry_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAgg",
        "dimensions": ["org_no", "industry_id", "data_time"],
        "rename": {"industry_id": "type_id"},
        "save_table": task_base_save_table,
        "is_valid": 1,
    },
    "step_all_ind_agg": {
        "type": "step_industry_agg",
        "module": "air_web.dw.dws_industry_15min",
        "class": "IndResAgg",
        "dimensions": ["org_no", "all_industry_id", "data_time"],
        "rename": {"all_industry_id": "type_id"},
        "save_table": task_base_save_table,
        "is_valid": 1,
    },
}
task_baseline = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.dws_15min",
        "class": "AggBase",
        "is_baseline": True,
    },
    "step_on7_type": {
        "type": "agg",
        "dimensions": ["on7", "type_id", "data_time"],
        "rename": {"on7": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_pare": {
        "type": "agg",
        "dimensions": ["on7", "pare_type_id", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_sort": {
        "type": "agg",
        "dimensions": ["on7", "type_code_sort", "data_time"],
        "rename": {"on7": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on7_all": {
        "type": "agg",
        "dimensions": ["on7", "highest_type_id", "data_time"],
        "rename": {"on7": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_type": {
        "type": "agg",
        "dimensions": ["on5", "type_id", "data_time"],
        "rename": {"on5": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "agg",
        "dimensions": ["on5", "pare_type_id", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_sort": {
        "type": "agg",
        "dimensions": ["on5", "type_code_sort", "data_time"],
        "rename": {"on5": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_on5_all": {
        "type": "agg",
        "dimensions": ["on5", "highest_type_id", "data_time"],
        "rename": {"on5": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_type": {
        "type": "agg",
        "dimensions": ["province_no", "type_id", "data_time"],
        "rename": {"province_no": "org_no"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "agg",
        "dimensions": ["province_no", "pare_type_id", "data_time"],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_sort": {
        "type": "agg",
        "dimensions": ["province_no", "type_code_sort", "data_time"],
        "rename": {"province_no": "org_no", "type_code_sort": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
    "step_pro_all": {
        "type": "agg",
        "dimensions": ["province_no", "highest_type_id", "data_time"],
        "rename": {"province_no": "org_no", "highest_type_id": "type_id"},
        "save_table": task_baseline_save_table,
        "save_es": 1,
        "is_valid": 1,
    },
}
task_day = {
    "step_org_day_stat": {
        "type": "org_day_stat",
        "module": "air_web.dw.dws_day",
        "class": "OrgDayStat",
        "source_table": "orgno_typeid_15min",
        "save_table": "orgno_typeid_day_stat",
        "is_valid": 1,
    },
    "step_city_report": {
        "type": "city_report",
        "module": "air_web.dw.city_report",
        "class": "AggCityReport",
        "source_table": "orgno_typeid_15min",
        "save_table": "city_report",
        "is_valid": 0,
    },
    "step_city_alert": {
        "type": "city_alert",
        "module": "air_web.dw.dws_day",
        "class": "GetCityAlert",
        "source_table": "",
        "save_table": "city_alert",
        "is_valid": 0,
    },
    "step_city_control": {
        "type": "city_control",
        "module": "air_web.dw.dws_day",
        "class": "CityControl",
        "source_table": "",
        "save_table": "city_control",
        "is_valid": 0,
    },
}
task_org_type_day = {
    "step_org_type_day_stat": {
        "type": "org_type_day_stat",
        "module": "air_web.dw.dws_day",
        "class": "OrgTypeDayStat",
        "source_table": "orgno_typeid_15min",
        "save_table": "orgno_typeid_ymd_stat",
        "is_valid": 1,
    }
}
task_cons_day = {
    "step_cons_day": {
        "type": "file",
        "module": "air_web.dw.cons_day_stat",
        "class": "AggConsDay",
        "save_table": "cons_day_stat",
        "is_valid": 1,
    }
}
task_cons_day_stat = {
    "task_cons_day_stat": {
        "type": "file",
        "module": "air_web.dw.cons_day_stat",
        "class": "ConsDayStat",
        "save_table": "cons_ymd_stat",
        "is_valid": 1,
    }
}
task_orderly = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.spec_cons_15min",
        "class": "SpecConsAggBase",
        "flag_field": "orderly_flag",
    },
    "step_on7_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on7", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on5", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "spec_cons_agg",
        "dimensions": [
            "province_no",
            "pare_type_id",
            "measure_code",
            "data_time",
        ],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": orderly_power_15min,
        "is_valid": 1,
    },
}
task_demand = {
    "step_base": {
        "type": "base_agg",
        "module": "air_web.dw.spec_cons_15min",
        "class": "SpecConsAggBase",
        "flag_field": "demand_flag",
    },
    "step_on7_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on7", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on7": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
    "step_on5_pare": {
        "type": "spec_cons_agg",
        "dimensions": ["on5", "pare_type_id", "measure_code", "data_time"],
        "rename": {"on5": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
    "step_pro_pare": {
        "type": "spec_cons_agg",
        "dimensions": [
            "province_no",
            "pare_type_id",
            "measure_code",
            "data_time",
        ],
        "rename": {"province_no": "org_no", "pare_type_id": "type_id"},
        "save_table": demand_power_15min,
        "is_valid": 1,
    },
}
task_cons_sync = {
    "step_cons_sync": {
        "type": "cons_sync",
        "module": "air_web.dw.cons_res_es_to_mysql",
        "class": "ConsResSync",
        "save_table": "aclr_res_power_96",
        "is_valid": 1,
    },
}

task_cons_sort = {
    "step_cons_sort": {
        "type": "file",
        "module": "air_web.scripts.hebei.cons_sort",
        "class": "ConsSort",
        "is_valid": 1,
    }
}


def parse_config(logger, task, start_time, end_time, timedelta, rewrite=False):
    task_id = task["task_id"]
    task_name = task["task_name"]
    pid = os.getpid()
    hostname = socket.gethostname()
    ip_addr = socket.gethostbyname(hostname)
    try:
        t1 = time.time()
        logger.info("开始处理task_id:{},start_time:{}".format(task_id, start_time))
        gen_task_log(
            task_id,
            task_name,
            StatusCode.PROCESSING,
            start_time,
            end_time,
            pid,
            ip_addr,
        )

        task_config = json.loads(task["config_text"])
        # 先处理最小颗粒度的聚合结果
        step_base_dict = task_config.pop("step_base", None)
        if step_base_dict:
            module = importlib.import_module(step_base_dict["module"])
            cls = getattr(module, step_base_dict["class"])
            step_base = cls(
                logger, task, start_time, end_time, "step_base", step_base_dict
            )
            base_df = step_base.main()

        # 处理配置中的其他聚合
        for step, step_dict in task_config.items():
            if not step_dict["is_valid"]:
                continue
            if step_dict["type"] == "agg":
                step = AggMain(
                    logger,
                    task,
                    start_time,
                    end_time,
                    step,
                    step_dict,
                    rewrite,
                    base_df,
                    step_base_dict["is_baseline"],
                )
            elif step_dict["type"] == "spec_cons_agg":
                step = SpecConsAggMain(
                    logger,
                    task,
                    start_time,
                    end_time,
                    step,
                    step_dict,
                    rewrite,
                    base_df,
                )
            elif step_dict["type"] == "step_industry_agg":
                module = importlib.import_module(step_dict["module"])
                cls = getattr(module, step_dict["class"])
                step = cls(
                    logger,
                    task,
                    start_time,
                    end_time,
                    step,
                    step_dict,
                    rewrite,
                    base_df,
                )
            else:
                module = importlib.import_module(step_dict["module"])
                cls = getattr(module, step_dict["class"])
                step = cls(
                    logger,
                    task,
                    start_time,
                    end_time,
                    step,
                    step_dict,
                    timedelta,
                    rewrite,
                )
            step.main()
        t2 = time.time()
        logger.info(
            "处理完成task_id:{},task_name:{},完成时间:{},start_time:{}".format(
                task_id, task_name, round(t2 - t1, 2), start_time
            )
        )
        gen_task_log(
            task_id,
            task_name,
            StatusCode.SUCCEEDED,
            start_time,
            end_time,
            pid,
            ip_addr,
        )
    except:
        logger.error(
            "处理失败task_id:{},start_time:{}".format(task_id, start_time)
        )
        logger.error(traceback.format_exc())
        gen_task_log(
            task_id,
            task_name,
            StatusCode.FAILED,
            start_time,
            end_time,
            pid,
            ip_addr,
        )


def proc_task(logger, task, start_time, end_time, timedelta, rewrite=False):
    p = Process(
        target=parse_config,
        args=(logger, task, start_time, end_time, timedelta, rewrite),
    )
    p.start()


if __name__ == "__main__":
    from air_web.dw.logger import init_log

    if not os.path.exists("./logs/"):
        os.mkdir("./logs/")
    log = init_log("./logs/")

    task_first = {
        "task_id": 1,
        "task_name": "task_first",
        "config_text": json.dumps(task_first),
    }
    task_base = {
        "task_id": 2,
        "task_name": "task_base",
        "config_text": json.dumps(task_base),
    }
    task_industry_agg = {
        "task_id": 3,
        "task_name": "task_industry_agg",
        "config_text": json.dumps(task_industry_agg),
    }
    task_baseline = {
        "task_id": 2,
        "task_name": "task_baseline",
        "config_text": json.dumps(task_baseline),
    }
    task_day = {
        "task_id": 4,
        "task_name": "task_day",
        "config_text": json.dumps(task_day),
    }
    task_org_type_day = {
        "task_id": 4,
        "task_name": "task_org_type_day",
        "config_text": json.dumps(task_org_type_day),
    }
    task_cons_day = {
        "task_id": 5,
        "task_name": "task_cons_day",
        "config_text": json.dumps(task_cons_day),
    }
    task_cons_day_stat = {
        "task_id": 5,
        "task_name": "task_cons_day_stat",
        "config_text": json.dumps(task_cons_day_stat),
    }
    task_orderly = {
        "task_id": 6,
        "task_name": "task_orderly",
        "config_text": json.dumps(task_orderly),
    }
    task_demand = {
        "task_id": 7,
        "task_name": "task_demand",
        "config_text": json.dumps(task_demand),
    }
    task_cons_sort = {
        "task_id": 8,
        "task_name": "task_sort",
        "config_text": json.dumps(task_cons_sort),
    }
    task_cons_sync = {
        "task_id": 8,
        "task_name": "task_cons_sync",
        "config_text": json.dumps(task_cons_sync),
    }

    # parse_config(
    #     log, task_first, "2019-01-01 00:00:00", "2019-01-02 00:00:00", "15 min"
    # )
    parse_config(
        log, task_base, "2024-04-10 00:00:00", "2024-04-11 00:00:00", "15 min"
    )
    # parse_config(
    #     log, task_industry_agg, "2020-06-28 00:00:00", "2020-06-29 00:00:00", "15 min"
    # )
    # parse_config(
    #     log, task_baseline, "2020-07-01 00:00:00", "2020-07-02 00:00:00", "15 min"
    # )
    # parse_config(
    #     log, task_day, "2020-06-28 00:00:00", "2020-06-29 00:00:00", "1 d"
    # )
    # parse_config(
    #     log,
    #     task_org_type_day,
    #     "2020-06-28 00:00:00",
    #     "2020-06-29 00:00:00",
    #     "1 d",
    # )
    # parse_config(
    #     log, task_cons_day, "2019-01-01 00:00:00", "2019-01-02 00:00:00", "1 d"
    # )
    # parse_config(
    #     log,
    #     task_cons_day_stat,
    #     "2020-06-28 00:00:00",
    #     "2020-06-29 00:00:00",
    #     "1 d",
    # )
    # parse_config(
    #     log, task_orderly, "2020-06-28 00:00:00", "2020-06-29 00:00:00", "1 d"
    # )
    # parse_config(
    #     log, task_demand, "2020-06-28 00:00:00", "2020-06-29 00:00:00", "1 d"
    # )
    # parse_config(
    #     log, task_cons_sort, "2020-12-28 00:00:00", "2020-06-29 00:00:00", "1 d"
    # )
    # parse_config(
    #     log, task_cons_sync, "2023-09-07 00:00:00", "2023-09-08 00:00:00", "1 d"
    # )

    print("===========================end")
